package shortlink.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import shortlink.dao.entity.*;
import shortlink.dao.mapper.*;
import shortlink.dto.biz.ShortLinkStatsRecordDTO;
import shortlink.mq.idempotent.MessageQueueIdempotentHandler;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

import static shortlink.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static shortlink.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(topic ="${rocketmq.producer.topic}",
consumerGroup ="${rocketmq.consumer.group}" )
public class ShortLinkStatsSaverocketmqConsumer implements RocketMQListener<Map<String,String>> {
    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;
    private final RedissonClient redissonClient;
    private final LinkgotoMapper linkgotoMapper;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocalStatsMapper linkLocalStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final ShortLinkMapper shortLinkMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;


    @Override
    public void onMessage(Map<String, String> producerMap) {
        String keys=producerMap.get("keys");
        if(messageQueueIdempotentHandler.isMessageBeingConsumed(keys)){
            // 判断当前的这个消息流程是否执行完成
            if (messageQueueIdempotentHandler.isAccomplish(keys)) {
                return;
            }
        }
        try{
            String fullShortUrl=producerMap.get("fullShortUrl");
            if(StrUtil.isNotBlank(fullShortUrl)){
                String gid=producerMap.get("gid");
                ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
                actualSaveShortLinkStats(fullShortUrl,gid,statsRecord);

            }
        }catch (Throwable ex){
            log.error("记录短链接监控消费异常", ex);
            try {
                messageQueueIdempotentHandler.delMessageProcessed(keys);
            } catch (Throwable remoteEx) {
                log.error("删除幂等标识错误", remoteEx);
            }
            throw ex;
        }
        messageQueueIdempotentHandler.setAccomplish(keys);

    }
    public void actualSaveShortLinkStats(String fullShortUrl,String gid,ShortLinkStatsRecordDTO statsRecord){
        fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());
        RReadWriteLock readWriteLock=redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY,fullShortUrl));
        RLock rlock=readWriteLock.readLock();
        rlock.lock();
        try{
            /*基础访问记录*/
            if (StrUtil.isBlank(gid)) {
                LambdaQueryWrapper<LinkgotoDO> queryWrapper = Wrappers.lambdaQuery(LinkgotoDO.class)
                        .eq(LinkgotoDO::getFullShortUrl, fullShortUrl);
                LinkgotoDO shortLinkGotoDO = linkgotoMapper.selectOne(queryWrapper);
                gid = shortLinkGotoDO.getGid();
            }
            LambdaQueryWrapper<LinkgotoDO> queryWrapper = Wrappers.lambdaQuery(LinkgotoDO.class)
                    .eq(LinkgotoDO::getFullShortUrl, statsRecord.getFullShortUrl());
            LinkgotoDO linkgotoDO=linkgotoMapper.selectOne(queryWrapper);
            Date currentDate=statsRecord.getCurrentDate();
            int hour= DateUtil.hour(currentDate,true);
            Week week=DateUtil.dayOfWeekEnum(currentDate);
            int weekValue=week.getIso8601Value();
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .date(currentDate)
                    .hour(hour)
                    .week(weekValue)
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .build();
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);
            /*访问地区*/
            Map<String , Object> localeParamMap=new HashMap<>();
            localeParamMap.put("key",statsLocaleAmapKey);
            localeParamMap.put("ip",statsRecord.getRemoteAddr());
            //获取高德地图的接口地址‘
            String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);
            JSONObject localeResultObj = JSON.parseObject(localeResultStr);
            String infoCode=localeResultObj.getString("infocode");
            String actualProvince="未知";
            String actualCity="未知";
            if(StrUtil.isNotBlank(infoCode)&&StrUtil.equals(infoCode,"10000")){
                String province=localeResultObj.getString("province");
                boolean unknowflag=StrUtil.equals(province,"[]");
                LinkLocaleStatsDO localeStatsDO = LinkLocaleStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .province(actualProvince = unknowflag ? actualProvince : province)
                        .city(actualCity = unknowflag ? actualCity : localeResultObj.getString("city"))
                        .cnt(1)
                        .adcode(unknowflag ? "未知" : localeResultObj.getString("adcode"))
                        .country("中国")
                        .date(currentDate)
                        .gid(gid)
                        .build();
                linkLocalStatsMapper.shortLinkLocaleState(localeStatsDO);
            }
            /*操作系统记录*/
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .os(statsRecord.getOs())
                    .cnt(1)
                    .date(currentDate)
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);
            /*浏览器记录*/
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())
                    .gid(gid)
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);
            /*设备记录*/
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())
                    .gid(gid)
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);

            /*网络记录*/
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())
                    .gid(gid)
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);

            /*访问日志*/
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .user(statsRecord.getUv())
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .ip(statsRecord.getRemoteAddr())
                    .browser(statsRecord.getBrowser())
                    .os(statsRecord.getOs())
                    .network(statsRecord.getNetwork())
                    .device(statsRecord.getDevice())
                    .local(StrUtil.join("-", "中国", actualProvince, actualCity))
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);
            /*短链接表更新*/
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(currentDate)
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);


        }finally {
            rlock.unlock();
        }

    }
}
